#include "RpcChannel.h"
#include "rpc.pb.h"
#include "../Buffer.h"
#include "../TcpConnection.h"

using namespace std::placeholders;

bool parseMessage(const TcpConnectionPtr& conn, Buffer* buf, std::string& msg)
{
    const char* crlf = buf->findCRLF();
    // const char* crlf = buf->find12();
    if (crlf)
    {
        msg = buf->retrieveAsUntilString(crlf);
        buf->retrieveUntil(crlf + 2);
        return true;
    }
    return false;
}

void sendRpcMessage(const TcpConnectionPtr& conn, const ::google::protobuf::Message& message)
{
    Buffer buf;
    // int byte_size = message.ByteSize();
    int byte_size = message.ByteSizeLong();
    // buf.ensureWritableBytes(byte_size);
    buf.ensureWritableBytes(byte_size + 2);


    std::string message_str;
    message.SerializeToString(&message_str);
    buf.append(message_str);
    buf.append("\r\n");
    // buf.append("12");
    conn->send(&buf);
}

RpcChannel::RpcChannel()
  : services_(NULL)
{
}

RpcChannel::RpcChannel(const TcpConnectionPtr& conn)
  : conn_(conn),
    services_(NULL)
{

}

RpcChannel::~RpcChannel()
{
    for (std::map<int64_t, OutstandingCall>::iterator it = outstandings_.begin(); it != outstandings_.end(); ++it)
    {
        OutstandingCall out = it->second;
        delete out.response;
        delete out.done;
    }
}

void RpcChannel::CallMethod(const ::google::protobuf::MethodDescriptor* method,
                            google::protobuf::RpcController* controller,
                            const ::google::protobuf::Message* request,
                            ::google::protobuf::Message* response,
                            ::google::protobuf::Closure* done)
{
    RpcMessage message;
    message.set_type(REQUEST);
    int64_t id = id_.incrementAndGet();
    message.set_id(id);
    message.set_service(method->service()->name());
    message.set_method(method->name());
    message.set_request(request->SerializeAsString());

    OutstandingCall out = { response, done };
    {
        std::lock_guard lock(mutex_);
        outstandings_[id] = out;
    }
    sendRpcMessage(conn_, message);
}

void RpcChannel::onMessage(const TcpConnectionPtr& conn,
                           Buffer* buf)
{
    while (buf->readable()) {
        RpcMessagePtr rpcmessage( prototype_->New() );
        // std::string msg = buf->retrieveAllAsString();
        std::string msg;
        if (!parseMessage(conn, buf, msg)) {
            std::cout << "wait for more data\n";
            return ;
        }
        rpcmessage->ParseFromString(msg);
        onRpcMessage(conn, rpcmessage);
    }
}

void RpcChannel::onRpcMessage(const TcpConnectionPtr& conn,
                              const RpcMessagePtr& messagePtr)
{
    RpcMessage& message = *messagePtr;

    if (message.type() == RESPONSE)
    {
        int64_t id = message.id();
        assert(message.has_response() || message.has_error());

        OutstandingCall out = { NULL, NULL };

        {
            std::lock_guard lock(mutex_);
            std::map<int64_t, OutstandingCall>::iterator it = outstandings_.find(id);
            if (it != outstandings_.end())
            {
                // 找到了 删除
                out = it->second;
                outstandings_.erase(it);
            }
        }

        if (out.response)
        {
            std::shared_ptr<google::protobuf::Message> d(out.response);
            if (message.has_response())
            {
                out.response->ParseFromString(message.response());
            }
            if (out.done)
            {
                out.done->Run();
            }
        }
    }

    // 服务端接受REQUEST
    else if (message.type() == REQUEST)
    {
        ErrorCode error = WRONG_PROTO;
        if (services_)
        {
            // cout << to_string(message.id()) + " message.service(): " + message.service() + "\n";
            std::map<std::string, google::protobuf::Service*>::const_iterator it = services_->find(message.service());
            // 存在service
            if (it != services_->end())
            {
                google::protobuf::Service* service = it->second;
                // std::cout << "service: " << it->first << std::endl;
                assert(service != NULL);
                const google::protobuf::ServiceDescriptor* desc = service->GetDescriptor();
                const google::protobuf::MethodDescriptor* method
                    = desc->FindMethodByName(message.method());
                if (method)
                {
                    std::shared_ptr<google::protobuf::Message> request(service->GetRequestPrototype(method).New());
                    if (request->ParseFromString(message.request()))
                    {
                        google::protobuf::Message* response = service->GetResponsePrototype(method).New();
                        int64_t id = message.id();
                        // std::cout << request->DebugString();
                        // 调用远程调用函数
                        service->CallMethod(method, nullptr, request.get(), response,
                                            google::protobuf::NewCallback(this, &RpcChannel::doneCallback, response, id));
                        error = NO_ERROR;
                    }
                    else
                    {
                        error = INVALID_REQUEST;
                    }
                }
                else
                {
                    error = NO_METHOD;
                }
            }
            else
            {
                error = NO_SERVICE;
            }
        }
        else
        {
            error = NO_SERVICE;
        }
        
        if (error != NO_ERROR)
        {
            std::cout << to_string(message.id()) + "  Message occur ERROR" << std::endl;
            switch (error)
            {
            case INVALID_REQUEST:
                cout << "INVALID_REQUEST\n";
                break;
            case NO_METHOD:
                cout << "NO_METHOD\n";
                break;
            case NO_SERVICE:
                cout << "NO_SERVICE\n";
                break;
            default:
                break;
            }
            RpcMessage response;
            response.set_type(RESPONSE);
            response.set_id(message.id());
            response.set_error(error);
            sendRpcMessage(conn_, response);
        }
    }
    else if (message.type() == ERROR) { }
}

void RpcChannel::doneCallback(::google::protobuf::Message* response, int64_t id)
{
    // std::cout << response->DebugString();
    std::shared_ptr<google::protobuf::Message> d(response);
    RpcMessage message;
    message.set_type(RESPONSE);
    message.set_id(id);
    message.set_response(response->SerializeAsString());
    sendRpcMessage(conn_, message);
}